package com.yifeng.repo.flink.data.transport.utils.ddl;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import org.apache.commons.collections.CollectionUtils;

import com.yifeng.repo.flink.data.transport.dto.AlterContent;
import com.yifeng.repo.flink.data.transport.dto.DdlContent;
import com.yifeng.repo.flink.data.transport.dto.DdlTypeEnums;
import com.yifeng.repo.flink.data.transport.dto.HistoryRecord;

/**
 * ddl数据解析Util类
 * @author lijing
 * @since 2023-04-14
 *
 */
public class DdlUtils {

	public static List<DdlContent> parseDdlContent(Map<String, Map<String,String>> renameMap,Map<String, List<String>> columnMap,HistoryRecord historyRecord) {
    	List<DdlContent> ddlContentList = new ArrayList<DdlContent>();
        //循环处理ColumnMap
        if(!columnMap.isEmpty()) {
        	for (Map.Entry<String, List<String>> entry : columnMap.entrySet()) {
        		DdlContent ddlContent = new DdlContent();
        		ddlContent.setDdlType(entry.getKey());
        		List<AlterContent> contentList = new ArrayList<AlterContent>();
        		List<String> columnList = entry.getValue();
                if(DdlTypeEnums.MODIFY.getType().equals(ddlContent.getDdlType())
        				|| DdlTypeEnums.ADD.getType().equals(ddlContent.getDdlType())) {
        			//根据列名从cdc返回的tableChange中查询相应的信息
        			if(CollectionUtils.isNotEmpty(columnList)) {
        				columnList.forEach(column->{
        					if(CollectionUtils.isNotEmpty(historyRecord.getTableChanges())
        					   && Objects.nonNull(historyRecord.getTableChanges().get(0).getTable())
        					   && CollectionUtils.isNotEmpty(historyRecord.getTableChanges().get(0).getTable().getColumns())) {
        						historyRecord.getTableChanges().get(0).getTable().getColumns().forEach(tableChangeColumn->{
        							if(column.equalsIgnoreCase(tableChangeColumn.getName())) {
        								AlterContent alterContent = new AlterContent();
        								alterContent.setName(column);
        								alterContent.setLength(tableChangeColumn.getLength());
        								alterContent.setScale(tableChangeColumn.getScale());
        								alterContent.setType(tableChangeColumn.getTypeName());
        								contentList.add(alterContent);
        							}
        						});
        					}
        				});
        			}
        		}
                if(DdlTypeEnums.DROP.getType().equals(ddlContent.getDdlType())) {
        			if(CollectionUtils.isNotEmpty(columnList)) {
        				columnList.forEach(column->{
        					AlterContent alterContent = new AlterContent();
        					alterContent.setName(column);
        					contentList.add(alterContent);
        				});
        			}
        		}
                ddlContent.setContentList(contentList);
                ddlContentList.add(ddlContent);
        	}
        }
        //循环处理RenameMap
        if(!renameMap.isEmpty()) {
        	for (Map.Entry<String, Map<String, String>> entry : renameMap.entrySet()) {
        		DdlContent ddlContent = new DdlContent();
        		ddlContent.setDdlType(entry.getKey());
        		ddlContent.setRenameContentMap(entry.getValue());
        		ddlContentList.add(ddlContent);
        	}
        }
        return ddlContentList;
    }
}
